package com.spring.example.rocket;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQListener;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;

/**
 * author: xiaochi
 * date: 2025/9/30
 */
@Slf4j
public abstract class AbstractListener<T> implements RocketMQListener<MqMessage<T>> {

    public void onMessage(MqMessage<T> message) {
        long startTime = System.currentTimeMillis();
        String messageId = message.getMessageId();
        String topic = message.getTopic();
        String tag = message.getTag();
        try {
            log.info("开始处理消息, messageId: {}, topic: {}, tag: {}", messageId, topic, tag);

            // 前置处理
           if (!preProcess(message)){
               return;
           }

            // 业务处理
            boolean success = handleMessage(message);

            if (success) {
                log.info("消息处理成功, messageId: {}, topic: {}, tag: {}, cost: {}ms", messageId, topic, tag, System.currentTimeMillis() - startTime);
            } else {
                log.warn("消息处理失败, messageId: {}, topic: {}, tag: {}", messageId, topic, tag);
                // 可在此处添加重试逻辑或发送到死信队列
                handleConsumeFailure(message, "业务处理返回失败");
            }

        } catch (Exception e) {
            log.error("消息处理异常, messageId: {}, topic: {}, tag: {}, error: {}", messageId, topic, tag, e);
            handleConsumeException(message, e.getMessage());
        }
    }

    /**
     * 前置处理（子类可重写）
     */
    protected boolean preProcess(MqMessage<T> message) {
        // 消息去重校验
        if (isMessageProcessed(message.getMessageId())) {
            log.info("消息已处理过，直接跳过, messageId: {}", message.getMessageId());
            return Boolean.FALSE;
        }
        // 参数校验
        if (message.getBody() == null) {
            // 消息体为空
            return Boolean.FALSE;
        }
        return Boolean.TRUE;
    }

    /**
     * 处理消息消费失败
     */
    protected void handleConsumeFailure(MqMessage<T> message, String errorMsg) {
        // 记录消费失败日志
        log.error("消息消费失败, messageId: {}, error: {}", message.getMessageId(), errorMsg);

        // 可在此处实现：
        // 1. 记录到数据库
        // 2. 发送告警
        // 3. 重试机制
        // 4. 发送到死信队列
    }

    /**
     * 处理消息消费失败
     */
    protected void handleConsumeException(MqMessage<T> message, String errorMsg) {
        // 记录消费失败日志
        log.error("消息消费失败, messageId: {}, error: {}", message.getMessageId(), errorMsg);

        // 可在此处实现：
        // 1. 记录到数据库
        // 2. 发送告警
        // 3. 重试机制
        // 4. 发送到死信队列
    }

    /**
     * 检查消息是否已处理（幂等性检查）
     */
    protected boolean isMessageProcessed(String messageId) {
        // 实现思路：
        // 1. 使用Redis记录已处理的消息ID，设置过期时间
        // 2. 使用数据库唯一索引
        // 3. 根据业务场景实现
        return false; // 默认返回false，子类根据实际情况实现
    }

    /**
     * 具体的消息处理方法（子类必须实现）
     */
    protected abstract boolean handleMessage(MqMessage<T> message);

    /**
     * 获取泛型数据类型(只能获取一级子类泛型，多级会报错)
     * @param object
     * @return
     */
    public Class<?> getGenericClass(Object object){
        //获取泛型的类型
        Type genericInterface = object.getClass().getGenericInterfaces()[0];
        ParameterizedType parameterizedType=(ParameterizedType) genericInterface;
        return (Class<?>)parameterizedType.getActualTypeArguments()[0];
    }
}
